package kafka;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class KafkaUtil {

	/**
	 * 创建一些topic
	 *
	 * @param brokerUrl kafka地址：localhost:9092
	 * @param topics    需要创建的topic，分区1副本1
	 * @return 创建结果
	 * @throws InterruptedException
	 */
	public static KafkaFuture<Void> createTopics(String brokerUrl, String[] topics) throws InterruptedException {

		Properties properties = new Properties();
		properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		AdminClient adminClient = AdminClient.create(properties);

		List<NewTopic> newTopics = new ArrayList<NewTopic>();

		for (String topic : topics) {

			NewTopic newTopic = new NewTopic(topic, 10, (short) 1);

			newTopics.add(newTopic);
		}

		CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);

		adminClient.close();

		Thread.sleep(5000);

		return createTopicsResult.all();
	}

	/**
	 * 查询所有的topic名称
	 *
	 * @param brokerUrl
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public static Set<String> getAllTopics(String brokerUrl) throws InterruptedException, ExecutionException {

		Properties properties = new Properties();
		properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		AdminClient adminClient = AdminClient.create(properties);

		ListTopicsResult topicRequestor = adminClient.listTopics();

		KafkaFuture<Set<String>> kafkaFuture = topicRequestor.names();

		adminClient.close();

		return kafkaFuture.get();
	}

	public static TopicDescription getTopicInfo(String brokerUrl, String topic) throws Exception {

		Properties properties = new Properties();
		properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		AdminClient adminClient = AdminClient.create(properties);

		List<String> topicList = new ArrayList<String>();

		topicList.add(topic);

		DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topicList);

		KafkaFuture<Map<String, TopicDescription>> kafkaFuture = describeTopicsResult.all();

		adminClient.close();

		return kafkaFuture.get().get(topic);
	}

	/**
	 * 删除topic
	 * 需要在kafka的配置文件中将auto.create.topics.enable设置为false、delete.topic.enable设置为true（没有相关配置项的话就直接添加）
	 *
	 * @param brokerUrl
	 * @param topics
	 * @return
	 * @throws InterruptedException
	 */
	public static KafkaFuture<Void> deleteTopics(String brokerUrl, String[] topics) throws InterruptedException {

		Properties properties = new Properties();
		properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		AdminClient adminClient = AdminClient.create(properties);

		DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(topics));

		adminClient.close();

		Thread.sleep(5000);

		return deleteTopicsResult.all();
	}

	public static void main(String[] args) throws Exception {

		String host = "49.234.13.92:9092";
//		String host = "127.0.0.1:9092";

//		System.out.println("清理TOPIC：" + deleteTopics(host, new String[] { "test" }));
		System.out.println("创建TOPIC：" + createTopics(host, new String[]{"perf_report_topic"}));
		System.out.println("查询TOPIC：" + getTopicInfo(host, "perf_report_topic"));

//		createTopics("127.0.0.1:9092", new String[] {"test55555"});

//		System.out.println(System.currentTimeMillis());
//		
//		Integer.valueOf(100);
	}
}
